<?php
   $rcf = new RdKafka\Conf();
	// $rcf->set('group.id', 'test');
   	$rk = new Rdkafka\Producer($rcf);
	$rk->setLogLevel(LOG_DEBUG);
	$rk->addBrokers("127.0.0.1:9092");   // 链接kafka服务器集群地址

	$cf = new RdKafka\TopicConf();
	$cf->set('offset.store.method', 'broker');
	$cf->set('auto.offset.reset', 'smallest');
	$topic = $rk->newTopic("test",$cf);   //   创建topic
	$message = "hello kafka " . date("Y-m-d H:i:s");  //消息内容
	 
	try {
		$topic->produce(RD_KAFKA_PARTITION_UA, 0, $message);  //写入消息
		sleep(2);
	} catch (\Exception $e) {
		echo $e->getMessage() . PHP_EOL;
	}


    echo "生产消息完毕".$message;




?>
